#!usr/bin/env python  
# -*- coding:utf-8 _*-
""" 
@author:robot
@file: mq03_rocketmq.py 
@version:
@time: 2022/03/07 
"""
import _thread
import time
import json
import random

from TapClientConnector.connector.tap_connector00_base import *
from TapUtil.utiltime.util_time import get_current_time

pk = 'pk'
mq_split = '<<<>>>'


class TapRocketMQConnector(TAPConnectorBase):

    def __init__(self, kwargs):
        self.pdk = kwargs.get('pdk')
        self.op = 'TAPDATA_SYNC_OP'
        self.sync_data = 'TAPDATA_SYNC_DATA'
        self.op_insert = 'INSERT'
        self.op_update = 'UPDATE'
        self.op_delete = 'DELETE'
        if self.pdk:
            self.op = 'mqOp'
            self.sync_data = 'sync_data'
            self.op_insert = 'insert'
            self.op_update = 'update'
            self.op_delete = 'delete'
        super(TapRocketMQConnector, self).__init__(kwargs)
        host = self.FORMAT_DB_INFO.get('host')
        port = self.FORMAT_DB_INFO.get('port')
        self.addr = f'{host}:{port}'
        self.consumer = random.choice('ABCDEFGHIJKLMNOPQRSTUVWXYZ')
        self.producer = 'Python-Producer'
        self.rocket_data = {}
        self._rocket_raw_data = []
        self.rocket_json_data = []
        self.init_timeout = 0
        self.timeout = 3600
        from rocketmq.client import Producer
        producer = Producer(self.producer)
        producer.set_namesrv_addr(self.addr)
        producer.start()
        producer.shutdown()

    def __rocket_transform(self, queue, pk=pk):
        self.rocket_json_data = []
        for column in self.rocket_data[queue]:
            op = column.get(self.op)
            data = column.get(self.sync_data)
            if op == self.op_insert:
                self.rocket_json_data.append(data)
            if op == self.op_update:
                for old in self.rocket_json_data:
                    try:
                        if old[pk] == data[pk]:
                            # for new_k, new_value in data.items():
                            #     old[new_k] = new_value
                            self.rocket_json_data.remove(old)
                            self.rocket_json_data.append(data)
                    except:
                        logger.warning('kafka 数据没有pk预设主键，查询整理结果无效')
            if op == self.op_delete:
                for old in self.rocket_json_data:
                    try:
                        if old[pk] == data[pk]:
                            self.rocket_json_data.remove(old)
                    except:
                        logger.warning('kafka 数据没有pk预设主键，查询整理结果无效')

    def __rocket_query(self, table):
        from rocketmq.client import PushConsumer
        consumer = PushConsumer(self.consumer)

        def start():
            consumer.set_namesrv_addr(self.addr)
            consumer.subscribe(table, self._callback)
            consumer.start()

        def close(consumer):
            for i in range(self.timeout + 1):
                time.sleep(1)
                self.init_timeout += 1
                if self.init_timeout == self.timeout:
                    consumer.shutdown()

        try:
            # # 默认等待线程
            # _thread.start_new_thread(close, (channel,))
            # 查询数据线程
            _thread.start_new_thread(start, ())
        except Exception:
            logger.fatal('thread error')
        close(consumer)
        return self._rocket_raw_data

    def _callback(self, msg):
        op = str(msg.get_property(self.op), 'UTF-8')
        _data = msg.body
        # import chardet
        # chardit1 = chardet.detect(msg.body)  # cf_r为要查看的编码
        # logger.fatal(chardit1['encoding'])
        data = json.loads(_data)
        logger.info({self.op: op})
        logger.info({self.sync_data: data})
        message = {
            self.op: op,
            self.sync_data: data
        }
        self._rocket_raw_data.append(message)

    def _callback2(self, msg):
        op = str(msg.get_property(self.op), 'UTF-8')
        _data = msg.body
        data = json.loads(_data)
        logger.info({self.op: op})
        logger.info({self.sync_data: data})
        message = {
            self.op: op,
            self.sync_data: data
        }
        self._rocket_raw_data.append(message)

    def _query(self, table):
        where = None
        if mq_split in table:
            _table = table.split(mq_split)
            table = _table[0]
            where = _table[1]
        self.rocket_data[table] = []
        res = self.__rocket_query(table)
        if not res:
            logger.fatal(f'rocketmq {table} consumer timeout {self.timeout} ms')
        self.rocket_data[table].extend(res)
        self.__rocket_transform(queue=table)
        res = self.rocket_json_data
        self.rocket_json_data = []
        self.rocket_data = {}
        if where:
            for one in res:
                if one[pk] == where:
                    return one
            return []
        return res

    def _exec(self, data):
        try:
            operate = data.get('operate')
            topic = data.get('queue')
            topic_data = data.get('queue_data')
        except:
            operate = 'insert'
            topic = data
            topic_data = {
                "pk": 1,
                "string": "string",
                "number": 1000,
                'boolean': True,
                'Array': [1, 2, 3],
                'object': {'test': 'test'}
            }
        from rocketmq.client import Producer, Message
        producer = Producer(self.producer)
        producer.set_namesrv_addr(self.addr)
        producer.start()

        json_data = json.dumps(topic_data)
        bytes_data = bytes(json_data, encoding='utf-8')
        msg = Message(topic)
        if operate == 'insert':
            # _queue_data.update({'TAPDATA_SYNC_OP': 'INSERT'})
            # _queue_data.update({'TAPDATA_SYNC_DATA': topic_data})
            # dict -> json -> utf-8
            msg.set_property(self.op, self.op_insert)
            msg.set_keys(get_current_time())
            msg.set_tags('explain')
            msg.set_body(bytes_data)
            producer.send_sync(msg)
        elif operate == 'update':
            msg.set_property(self.op, self.op_update)
            msg.set_keys(get_current_time())
            msg.set_tags('explain')
            msg.set_body(bytes_data)
            producer.send_sync(msg)
        elif operate == 'delete':
            msg.set_property(self.op, self.op_delete)
            msg.set_keys(get_current_time())
            msg.set_tags('explain')
            msg.set_body(bytes_data)
            producer.send_sync(msg)
        producer.shutdown()

    @generate_logger
    def generate_select(self, table=None, where=None, order=''):
        logger.info('consumer time is 5s')
        self.timeout = 5
        if table:
            self.CURRENT_TABLE = table
        if where:
            return f'{self.CURRENT_TABLE}{mq_split}{where.get(pk)}'
        return self.CURRENT_TABLE

    @generate_logger
    def generate_insert(self, insert, table=None):
        if table:
            self.CURRENT_TABLE = table
        return {
            'operate': 'insert',
            'queue_data': insert,
            'queue': self.CURRENT_TABLE
        }

    @generate_logger
    def generate_update(self, update, where, table=None):
        if table:
            self.CURRENT_TABLE = table
        return {
            'operate': 'update',
            'data': (where, update),
            'queue_data': self.CURRENT_TABLE
        }

    @generate_logger
    def generate_delete(self, where, table=None):
        if table:
            self.CURRENT_TABLE = table
        return {
            'operate': 'delete',
            'data': where,
            'queue_data': self.CURRENT_TABLE
        }

    @generate_logger
    def generate_create(self, table_column, table=None):
        if table:
            self.CURRENT_TABLE = table
        return self.generate_insert(table_column, table=self.CURRENT_TABLE)

    @generate_logger
    def generate_drop(self, table=None):
        if table:
            self.CURRENT_TABLE = table
        return {
            'operate': 'drop',
            'queue_data': [self.CURRENT_TABLE]
        }


if __name__ == '__main__':
    T = TapRocketMQConnector(MQ.Rocket_only_for_auto)
    T.timeout = 10
    print(T._query('auto_dummy'))
